package chapter03
import org.apache.log4j.{Level, Logger}
import org.apache.spark.{SparkConf, SparkContext}
object Test36_transform {
  def main(args: Array[String]): Unit = {
    Logger.getLogger("org.apache.spark").setLevel(Level.WARN)
    val conf = new SparkConf().setMaster("local[*]").setAppName("transform")
    val sc = new SparkContext(conf)
    val value = sc.textFile("input/user_visit_action.txt")
    //切分文件
    val value1 = value.map(e => e.split("_"))
    //统计每个页面出现的次数
    val count = value1.map(e => e(3))
      .map(e => (e, 1))
      .reduceByKey(_ + _)
    println(count.take(3).toList)
    //统计每一个session访问的页面
    val value2 = value1.groupBy(e => e(2))
      .mapValues(e => e.toList.sortBy(f => f(4)))
      .mapValues(e => e.map(f => f(3)))
    println(value2.take(2).toList)
    val value3 = value2
      .mapValues(e => e.zip(e.tail))
    println(value3.take(2).toList)
    //计算所有组合的个数
    val value4 = value3.flatMap(e => e._2)
      .map(e => (e, 1))
      .reduceByKey(_ + _)
    println(value4.take(2).toList)
    //计算转化率 组合中第一个页面出现的次数/第一个元素出现的总次数
    val map = count.collect().toMap
    val value5 = value4
      .map(e => (e._1, e._2.toDouble / map.getOrElse(e._1._1, 0)))
    println(value5.take(5).toList)
  }
}
